package demoflink.sink;

import demoflink.entity.WaterSensor;
import demoflink.function.WaterSensorMapFunction;
import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

public class SinkCustom {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);
        DataStreamSource<String> source = env.socketTextStream("192.168.101.121", 7777);
        SingleOutputStreamOperator<WaterSensor> operator = source.map(new WaterSensorMapFunction());
        operator.addSink(new MySink());
        env.execute();
    }
    public static class MySink extends RichSinkFunction<WaterSensor> {


        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);

            //连接池复用
        }

        @Override
        public void close() throws Exception {
            super.close();
            //关闭连接池
        }

        @Override
        public void invoke(WaterSensor value, Context context) throws Exception {
            super.invoke(value, context);
            //写出逻辑
        }

        @Override
        public void writeWatermark(Watermark watermark) throws Exception {
            super.writeWatermark(watermark);
        }

        @Override
        public void finish() throws Exception {
            super.finish();
        }
    }
}
